package drds.binlog.client.impl;

import com.google.protobuf.ByteString;
import drds.binlog.client.Connector;
import drds.binlog.client.impl.running.ClientRunningData;
import drds.binlog.client.impl.running.ClientRunningListener;
import drds.binlog.client.impl.running.ClientRunningMonitor;
import drds.binlog.common.position.ClientInfo;
import drds.binlog.common.utils.AddressUtils;
import drds.binlog.common.utils.BooleanMutex;
import drds.binlog.common.zookeeper.ZooKeeperClient;
import drds.binlog.protocol.Message;
import drds.binlog.protocol.entry_protocol_proto.Entry;
import drds.binlog.protocol.exception.ClientException;
import drds.binlog.protocol.protocol_proto.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.ByteOrder;
import java.nio.channels.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * 基于{@linkplain canalserverwithnetty}定义的网络协议接口，对于canal数据进行get/rollback/ack等操作
 */
public class SimpleConnector implements Connector
{

    private static final Logger logger = LoggerFactory.getLogger(SimpleConnector.class);
    private final ByteBuffer readHeader = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
    private final ByteBuffer writeHeader = ByteBuffer.allocate(4).order(ByteOrder.BIG_ENDIAN);
    private SocketAddress socketAddress;
    private String username;
    private String password;
    private int soTimeout = 60000;                                              // milliseconds
    private int idleTimeout = 60 * 60 * 1000;                                     // client和server之间的空闲链接超时的时间,默认为1小时
    private String filter;                                                                     // 记录上一次的filter提交值,便于自动重试时提交
    private SocketChannel socketChannel;
    private ReadableByteChannel readableByteChannel;
    private WritableByteChannel writableByteChannel;
    private List<Compression> supportedCompressions = new ArrayList<Compression>();
    private ClientInfo clientInfo;
    private ClientRunningMonitor clientRunningMonitor;                                                             // 运行控制
    private ZooKeeperClient zooKeeperClient;
    private BooleanMutex mutex = new BooleanMutex(false);
    private volatile boolean connected = false;                                              // 代表connected是否已正常执行，因为有HA，不代表在工作中
    private boolean rollbackOnConnect = true;                                               // 是否在connect链接成功后，自动执行rollback操作
    private boolean rollbackOnDisConnect = false;                                              // 是否在connect链接成功后，自动执行rollback操作
    private boolean lazyParseEntry = false;                                              // 是否自动化解析Entry对象,如果考虑最大化性能可以延后解析
    // 读写数据分别使用不同的锁进行控制，减小锁粒度,读也需要排他锁，并发度容易造成数据包混乱，反序列化失败
    private Object readDataLock = new Object();
    private Object writeDataLock = new Object();

    private volatile boolean running = false;

    public SimpleConnector(SocketAddress socketAddress, String username, String password, String destination)
    {
        this(socketAddress, username, password, destination, 60000, 60 * 60 * 1000);
    }

    public SimpleConnector(SocketAddress socketAddress, String username, String password, String destination,
                           int soTimeout)
    {
        this(socketAddress, username, password, destination, soTimeout, 60 * 60 * 1000);
    }

    public SimpleConnector(SocketAddress socketAddress, String username, String password, String destination,
                           int soTimeout, int idleTimeout)
    {
        this.socketAddress = socketAddress;
        this.username = username;
        this.password = password;
        this.soTimeout = soTimeout;
        this.idleTimeout = idleTimeout;
        this.clientInfo = new ClientInfo(destination, (short) 1001);
    }

    public void connect() throws ClientException
    {
        if (connected)
        {
            return;
        }

        if (clientRunningMonitor != null)
        {
            if (!clientRunningMonitor.isRunning())
            {
                clientRunningMonitor.start();
            }
        } else
        {
            waitClientRunning();
            if (!running)
            {
                return;
            }
            doConnect();
            if (filter != null)
            { // 如果存在条件，说明是自动切换，基于上一次的条件订阅一次
                subscribe(filter);
            }
            if (rollbackOnConnect)
            {
                rollback();
            }
        }

        connected = true;
    }

    public void disconnect() throws ClientException
    {
        if (rollbackOnDisConnect && socketChannel.isConnected())
        {
            rollback();
        }

        connected = false;
        if (clientRunningMonitor != null)
        {
            if (clientRunningMonitor.isRunning())
            {
                clientRunningMonitor.stop();
            }
        } else
        {
            doDisconnnect();
        }
    }

    private InetSocketAddress doConnect() throws ClientException
    {
        try
        {
            socketChannel = SocketChannel.open();
            socketChannel.socket().setSoTimeout(soTimeout);
            SocketAddress socketAddress = getSocketAddress();
            if (socketAddress == null)
            {
                socketAddress = getNextAddress();
            }
            socketChannel.connect(socketAddress);
            readableByteChannel = Channels.newChannel(socketChannel.socket().getInputStream());
            writableByteChannel = Channels.newChannel(socketChannel.socket().getOutputStream());
            Packet packet1 = Packet.parseFrom(readNextPacket());
            if (packet1.getVersion() != 1)
            {
                throw new ClientException("unsupported version at this client.");
            }

            if (packet1.getType() != PacketType.handshake)
            {
                throw new ClientException("expect handshake but found other type.");
            }
            //
            Handshake handshake = Handshake.parseFrom(packet1.getBody());
            supportedCompressions.add(handshake.getSupportedCompressions());
            //
            ClientAuth clientAuth = ClientAuth.newBuilder()
                    .setUsername(username != null ? username : "")
                    .setPassword(ByteString.copyFromUtf8(password != null ? password : ""))
                    .setNetReadTimeout(idleTimeout)
                    .setNetWriteTimeout(idleTimeout)
                    .build();
            writeWithHeader(Packet.newBuilder()
                    .setType(PacketType.clientauthentication)
                    .setBody(clientAuth.toByteString())
                    .build()
                    .toByteArray());
            //
            Packet packet = Packet.parseFrom(readNextPacket());
            if (packet.getType() != PacketType.ack)
            {
                throw new ClientException("unexpected packet type when ack is expected");
            }

            Ack ack = Ack.parseFrom(packet.getBody());
            if (ack.getErrorCode() > 0)
            {
                throw new ClientException("something goes wrong when doing authentication: "
                        + ack.getErrorMessage());
            }

            connected = true;
            return new InetSocketAddress(socketChannel.socket().getLocalAddress(), socketChannel.socket().getLocalPort());
        } catch (IOException e)
        {
            throw new ClientException(e);
        }
    }

    private void doDisconnnect() throws ClientException
    {
        if (readableByteChannel != null)
        {
            quietlyClose(readableByteChannel);
            readableByteChannel = null;
        }
        if (writableByteChannel != null)
        {
            quietlyClose(writableByteChannel);
            writableByteChannel = null;
        }
        if (socketChannel != null)
        {
            quietlyClose(socketChannel);
            socketChannel = null;
        }
    }

    private void quietlyClose(Channel channel)
    {
        try
        {
            channel.close();
        } catch (IOException e)
        {
            logger.warn("exception on closing socketChannel:{} \n {}", channel, e);
        }
    }

    public void subscribe() throws ClientException
    {
        subscribe(""); // 传递空字符即可
    }

    public void subscribe(String filter) throws ClientException
    {
        waitClientRunning();
        if (!running)
        {
            return;
        }
        try
        {
            writeWithHeader(Packet.newBuilder()
                    .setType(PacketType.subscription)
                    .setBody(Sub.newBuilder()
                            .setDestination(clientInfo.getDestination())
                            .setClientId(String.valueOf(clientInfo.getClientId()))
                            .setFilter(filter != null ? filter : "")
                            .build()
                            .toByteString())
                    .build()
                    .toByteArray());
            //
            Packet p = Packet.parseFrom(readNextPacket());
            Ack ack = Ack.parseFrom(p.getBody());
            if (ack.getErrorCode() > 0)
            {
                throw new ClientException("failed to subscribe with reason: " + ack.getErrorMessage());
            }

            clientInfo.setFilter(filter);
        } catch (IOException e)
        {
            throw new ClientException(e);
        }
    }

    public void unsubscribe() throws ClientException
    {
        waitClientRunning();
        if (!running)
        {
            return;
        }
        try
        {
            writeWithHeader(Packet.newBuilder()
                    .setType(PacketType.unsubscription)
                    .setBody(Unsub.newBuilder()
                            .setDestination(clientInfo.getDestination())
                            .setClientId(String.valueOf(clientInfo.getClientId()))
                            .build()
                            .toByteString())
                    .build()
                    .toByteArray());
            //
            Packet p = Packet.parseFrom(readNextPacket());
            Ack ack = Ack.parseFrom(p.getBody());
            if (ack.getErrorCode() > 0)
            {
                throw new ClientException("failed to unSubscribe with reason: " + ack.getErrorMessage());
            }
        } catch (IOException e)
        {
            throw new ClientException(e);
        }
    }

    public Message get(int batchSize) throws ClientException
    {
        return get(batchSize, null, null);
    }

    public Message get(int batchSize, Long timeout, TimeUnit unit) throws ClientException
    {
        Message message = getWithoutAck(batchSize, timeout, unit);
        ack(message.getId());
        return message;
    }

    public Message getWithoutAck(int batchSize) throws ClientException
    {
        return getWithoutAck(batchSize, null, null);
    }

    public Message getWithoutAck(int batchSize, Long timeout, TimeUnit unit) throws ClientException
    {
        waitClientRunning();
        if (!running)
        {
            return null;
        }
        try
        {
            int size = (batchSize <= 0) ? 1000 : batchSize;
            long time = (timeout == null || timeout < 0) ? -1 : timeout; // -1代表不做timeout控制
            if (unit == null)
            {
                unit = TimeUnit.MILLISECONDS;
            }

            writeWithHeader(Packet.newBuilder()
                    .setType(PacketType.get)
                    .setBody(Get.newBuilder()
                            .setAutoAck(false)
                            .setDestination(clientInfo.getDestination())
                            .setClientId(String.valueOf(clientInfo.getClientId()))
                            .setFetchSize(size)
                            .setTimeout(time)
                            .setUnit(unit.ordinal())
                            .build()
                            .toByteString())
                    .build()
                    .toByteArray());
            return receiveMessages();
        } catch (IOException e)
        {
            throw new ClientException(e);
        }
    }

    private Message receiveMessages() throws IOException
    {
        byte[] bytes = readNextPacket();
        Packet packet = Packet.parseFrom(bytes);
        switch (packet.getType())
        {
            case messages:
            {
                if (!packet.getCompression().equals(Compression.none))
                {
                    throw new ClientException("compression is not supported in this connector");
                }

                Messages messages = Messages.parseFrom(packet.getBody());
                Message message = new Message(messages.getBatchId());
                if (lazyParseEntry)
                {
                    // byteString
                    message.setRawEntryList(messages.getMessagesList());
                } else
                {
                    for (ByteString byteString : messages.getMessagesList())
                    {
                        message.addEntry(Entry.parseFrom(byteString));
                    }
                }
                return message;
            }
            case ack:
            {
                Ack ack = Ack.parseFrom(packet.getBody());
                throw new ClientException("something goes wrong with reason: " + ack.getErrorMessage());
            }
            default:
            {
                throw new ClientException("unexpected packet type: " + packet.getType());
            }
        }
    }

    public void ack(long batchId) throws ClientException
    {
        waitClientRunning();
        if (!running)
        {
            return;
        }
        ClientAck ca = ClientAck.newBuilder()
                .setDestination(clientInfo.getDestination())
                .setClientId(String.valueOf(clientInfo.getClientId()))
                .setBatchId(batchId)
                .build();
        try
        {
            writeWithHeader(Packet.newBuilder()
                    .setType(PacketType.clientack)
                    .setBody(ca.toByteString())
                    .build()
                    .toByteArray());
        } catch (IOException e)
        {
            throw new ClientException(e);
        }
    }

    public void rollback(long batchId) throws ClientException
    {
        waitClientRunning();
        ClientRollback ca = ClientRollback.newBuilder()
                .setDestination(clientInfo.getDestination())
                .setClientId(String.valueOf(clientInfo.getClientId()))
                .setBatchId(batchId)
                .build();
        try
        {
            writeWithHeader(Packet.newBuilder()
                    .setType(PacketType.clientrollback)
                    .setBody(ca.toByteString())
                    .build()
                    .toByteArray());
        } catch (IOException e)
        {
            throw new ClientException(e);
        }
    }

    public void rollback() throws ClientException
    {
        waitClientRunning();
        rollback(0);// 0代笔未设置
    }

    // ==================== helper method ====================

    private void writeWithHeader(byte[] body) throws IOException
    {
        writeWithHeader(writableByteChannel, body);
    }

    private byte[] readNextPacket() throws IOException
    {
        return readNextPacket(readableByteChannel);
    }

    private void writeWithHeader(WritableByteChannel channel, byte[] body) throws IOException
    {
        synchronized (writeDataLock)
        {
            writeHeader.clear();
            writeHeader.putInt(body.length);
            writeHeader.flip();
            channel.write(writeHeader);
            channel.write(ByteBuffer.wrap(body));
        }
    }

    private byte[] readNextPacket(ReadableByteChannel readableByteChannel) throws IOException
    {
        synchronized (readDataLock)
        {
            readHeader.clear();
            read(readableByteChannel, readHeader);
            int bodyLen = readHeader.getInt(0);
            ByteBuffer byteBuffer = ByteBuffer.allocate(bodyLen).order(ByteOrder.BIG_ENDIAN);
            read(readableByteChannel, byteBuffer);
            return byteBuffer.array();
        }
    }

    private void read(ReadableByteChannel readableByteChannel, ByteBuffer byteBuffer) throws IOException
    {
        while (byteBuffer.hasRemaining())
        {
            int r = readableByteChannel.read(byteBuffer);
            if (r == -1)
            {
                throw new IOException("end of stream when reading header");
            }
        }
    }

    private synchronized void initClientRunningMonitor(ClientInfo clientInfo)
    {
        if (zooKeeperClient != null && clientInfo != null && clientRunningMonitor == null)
        {
            ClientRunningData clientData = new ClientRunningData();
            clientData.setClientId(clientInfo.getClientId());
            clientData.setAddress(AddressUtils.getHostIp());

            clientRunningMonitor = new ClientRunningMonitor();
            clientRunningMonitor.setDestination(clientInfo.getDestination());
            clientRunningMonitor.setZooKeeperClient(zooKeeperClient);
            clientRunningMonitor.setClientData(clientData);
            clientRunningMonitor.setListener(new ClientRunningListener()
            {

                public InetSocketAddress processActiveEnter()
                {
                    InetSocketAddress address = doConnect();
                    mutex.set(true);
                    if (filter != null)
                    { // 如果存在条件，说明是自动切换，基于上一次的条件订阅一次
                        subscribe(filter);
                    }

                    if (rollbackOnConnect)
                    {
                        rollback();
                    }

                    return address;
                }

                public void processActiveExit()
                {
                    mutex.set(false);
                    doDisconnnect();
                }

            });
        }
    }

    private void waitClientRunning()
    {
        try
        {
            if (zooKeeperClient != null)
            {
                if (!connected)
                {// 未调用connect
                    throw new ClientException("should connect first");
                }

                running = true;
                mutex.get();// 阻塞等待
            } else
            {
                // 单机模式直接设置为running
                running = true;
            }
        } catch (InterruptedException e)
        {
            Thread.currentThread().interrupt();
            throw new ClientException(e);
        }
    }

    public boolean checkValid()
    {
        if (zooKeeperClient != null)
        {
            return mutex.state();
        } else
        {
            return true;// 默认都放过
        }
    }

    public SocketAddress getNextAddress()
    {
        return null;
    }

    public SocketAddress getSocketAddress()
    {
        return socketAddress;
    }

    public String getUsername()
    {
        return username;
    }

    public String getPassword()
    {
        return password;
    }

    public int getSoTimeout()
    {
        return soTimeout;
    }

    public void setSoTimeout(int soTimeout)
    {
        this.soTimeout = soTimeout;
    }

    public int getIdleTimeout()
    {
        return idleTimeout;
    }

    public void setIdleTimeout(int idleTimeout)
    {
        this.idleTimeout = idleTimeout;
    }

    public void setZooKeeperClient(ZooKeeperClient zooKeeperClient)
    {
        this.zooKeeperClient = zooKeeperClient;
        initClientRunningMonitor(this.clientInfo);
    }

    public void setRollbackOnConnect(boolean rollbackOnConnect)
    {
        this.rollbackOnConnect = rollbackOnConnect;
    }

    public void setRollbackOnDisConnect(boolean rollbackOnDisConnect)
    {
        this.rollbackOnDisConnect = rollbackOnDisConnect;
    }

    public void setFilter(String filter)
    {
        this.filter = filter;
    }

    public boolean isLazyParseEntry()
    {
        return lazyParseEntry;
    }

    public void setLazyParseEntry(boolean lazyParseEntry)
    {
        this.lazyParseEntry = lazyParseEntry;
    }

    public void stopRunning()
    {
        if (running)
        {
            running = false; // 设置为非running状态
            if (!mutex.state())
            {
                mutex.set(true); // 中断阻塞
            }
        }
    }

}
